In this short guide, we will build a simple parrot agent together using Python. The parrot agent will simply repeat everything you send him and a small 🦜 emoji in front of the return. We will create the Open Floor Protocol-compliant agent with the help of the openfloor Python package.
Initial Setup
First, let's set up our project by creating the project folder and installing the required packages:
mkdir parrot-agent cd parrot-agent python -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate
Create a requirements.txt
file with the following content:
--extra-index-url https://test.pypi.org/simple/ events==0.5 jsonpath-ng>=1.5.0 openfloor flask
Then install the dependencies:
pip install -r requirements.txt
Now that the basic setup is done, let us start coding together!
Step 1: Building the Parrot Agent Class
Let's create our main agent file. Create a new file parrot_agent.py
, this will contain the main logic of our agent.
Step 1.1: Add the imports
Let's start with importing everything we need from the openfloor
package, add them at the top of your parrot_agent.py
file:
from openfloor import ( BotAgent, Manifest, Identification, Capability, SupportedLayers, UtteranceEvent, Envelope, DialogEvent, TextFeature, To, Sender, PublishManifestsEvent, Parameters )
Why these imports?
-
BotAgent
- The base class we'll extend -
Manifest
- To define our agent's capabilities and identification -
UtteranceEvent
- The type of event we'll handle for text messages -
Envelope
- Container for Open Floor messages -
DialogEvent
andTextFeature
- To create text responses -
To
andSender
- For message addressing
Step 1.2: Start the ParrotAgent class
Now let's start creating our ParrotAgent
class by extending the BotAgent
:
class ParrotAgent(BotAgent): """ ParrotAgent - A simple agent that echoes back whatever it receives Extends BotAgent to provide parrot functionality """ def __init__(self, manifest: Manifest): super().__init__(manifest) print(f"🦜 Parrot Agent initialized with speaker URI: {self.speakerUri}")
What we just did:
- Created a class that extends
BotAgent
- Added a constructor that takes a manifest and passes it to the parent class
- The manifest will define what our agent can do
Step 1.3: Override the utterance handler
The BotAgent
class provides a default bot_on_utterance
method that we need to override. This is where the magic happens:
def bot_on_utterance(self, event: UtteranceEvent, in_envelope: Envelope, out_envelope: Envelope) -> None: """ Override the utterance handler to provide parrot functionality """ print("🦜 Processing utterance event") try: # Extract the dialog event from the utterance parameters dialog_event_data = event.parameters.get("dialogEvent") if not dialog_event_data: self._send_error_response("*chirp* I didn't receive a valid dialog event!", out_envelope) return # Convert to DialogEvent object if it's a dictionary if isinstance(dialog_event_data, dict): dialog_event = DialogEvent.from_dict(dialog_event_data) elif isinstance(dialog_event_data, DialogEvent): dialog_event = dialog_event_data else: self._send_error_response("*chirp* I didn't receive a valid dialog event!", out_envelope) return # Check if there's a text feature text_feature = dialog_event.features.get("text") if not text_feature or not hasattr(text_feature, 'tokens') or not text_feature.tokens: self._send_error_response("*chirp* I can only repeat text messages!", out_envelope) return # Extract the original text from tokens original_text = "".join(token.value for token in text_feature.tokens if token.value) # Create parrot response with emoji prefix parrot_text = f"🦜 {original_text}" # Create the response dialog event response_dialog = DialogEvent( speakerUri=self.speakerUri, features={"text": TextFeature(values=[parrot_text])} ) # Create and add the utterance event to the response response_utterance = UtteranceEvent( dialogEvent=response_dialog, to=To(speakerUri=in_envelope.sender.speakerUri) ) out_envelope.events.append(response_utterance) print(f"🦜 Echoing back: {parrot_text}") except Exception as error: print(f"🦜 Error in parrot utterance handling: {error}") self._send_error_response("*confused chirp* Something went wrong while trying to repeat that!", out_envelope)
The parroting logic:
- Extract the dialog event from the utterance parameters
- Get the text feature and extract text from tokens
- Add the 🦜 emoji prefix
- Create a dialog event response
- Send it back to the original sender
Step 1.4: Add helper methods
Let's add the helper methods we called in bot_on_utterance
:
def _send_error_response(self, message: str, out_envelope: Envelope) -> None: """Helper method to send error responses""" error_dialog = DialogEvent( speakerUri=self.speakerUri, features={"text": TextFeature(values=[message])} ) error_utterance = UtteranceEvent( dialogEvent=error_dialog ) out_envelope.events.append(error_utterance)
Step 1.5: Override the manifest handler
We also need to handle manifest requests properly:
def bot_on_get_manifests(self, event, in_envelope: Envelope, out_envelope: Envelope) -> None: """ Handle manifest requests by sending our capabilities """ print("🦜 Sending manifest information") # Create the publish manifests response publish_event = PublishManifestsEvent( parameters=Parameters({ "servicingManifests": [self._manifest], "discoveryManifests": [] }), to=To(speakerUri=in_envelope.sender.speakerUri) ) out_envelope.events.append(publish_event)
Step 1.6: Add the factory function
After the class, add this factory function with the default configuration:
def create_parrot_agent( speaker_uri: str, service_url: str, name: str = "Parrot Agent", organization: str = "OpenFloor Demo", description: str = "A simple parrot agent that echoes back messages with a 🦜 emoji" ) -> ParrotAgent: """ Factory function to create a ParrotAgent with default configuration """ # Create the identification identification = Identification( speakerUri=speaker_uri, serviceUrl=service_url, organization=organization, conversationalName=name, synopsis=description ) # Create the capabilities capability = Capability( keyphrases=['echo', 'repeat', 'parrot', 'say'], descriptions=[ 'Echoes back any text message with a 🦜 emoji', 'Repeats user input verbatim', 'Simple text mirroring functionality' ], supportedLayers=SupportedLayers( input=["text"], output=["text"] ) ) # Create the manifest manifest = Manifest( identification=identification, capabilities=[capability] ) return ParrotAgent(manifest)
What this factory does:
- Takes configuration options with some defaults
- Creates an identification object that describes our agent
- Defines capabilities that tell others what our agent can do
- Creates a manifest that combines identification and capabilities
- Returns a new ParrotAgent instance
Step 2: Building the Flask Server
The agent itself is done, but how to talk to it? We need to build our Flask server for this, so start with creating a server.py
file.
Step 2.1: Add imports
Add these imports at the top:
from flask import Flask, request, jsonify from parrot_agent import create_parrot_agent from openfloor import Envelope, Payload import json import os
Step 2.2: Create the Flask app
app = Flask(__name__) # Configure CORS for specific origin @app.after_request def after_request(response): allowed_origin = 'http://127.0.0.1:4000' origin = request.headers.get('Origin') if origin == allowed_origin: response.headers.add('Access-Control-Allow-Origin', allowed_origin) response.headers.add('Access-Control-Allow-Methods', 'POST, OPTIONS') response.headers.add('Access-Control-Allow-Headers', 'Content-Type') return response @app.route('/', methods=['OPTIONS']) def handle_options(): """Handle preflight OPTIONS requests""" return '', 200
Why this CORS setup?
- Only allows requests from the specific domain
- Handles preflight
OPTIONS
requests - Restricts to
POST
methods and theContent-Type
header
Step 2.3: Create the agent instance
Now we need to create our parrot by using the factory function create_parrot_agent
.
# Create the parrot agent instance parrot_agent = create_parrot_agent( speaker_uri='tag:openfloor-demo.com,2025:parrot-agent', service_url=os.environ.get('SERVICE_URL', 'http://localhost:8080/'), name='Polly the Parrot', organization='OpenFloor Demo Corp', description='A friendly parrot that repeats everything you say!' ) print(f"🦜 Parrot agent created: {parrot_agent.speakerUri}")
Step 2.4: Build the main endpoint step by step
Now we have the agent and the Flask app, but the most important part is still missing, and that's our endpoint:
@app.route('/', methods=['POST']) def handle_openfloor_message(): """Main Open Floor Protocol endpoint""" try: print(f"🦜 Received request: {json.dumps(request.json, indent=2)}") # Validate the incoming payload if not request.json: return jsonify({ 'error': 'Invalid JSON payload' }), 400 # Parse the payload try: if 'openFloor' in request.json: # Direct payload format payload = Payload.from_dict(request.json) incoming_envelope = payload.openFloor else: # Direct envelope format incoming_envelope = Envelope.from_dict(request.json) except Exception as parse_error: print(f"🦜 Parsing error: {parse_error}") return jsonify({ 'error': 'Invalid OpenFloor payload format', 'details': str(parse_error) }), 400 print(f"🦜 Processing envelope from: {incoming_envelope.sender.speakerUri}") # Process the envelope through the parrot agent outgoing_envelope = parrot_agent.process_envelope(incoming_envelope) # Create response payload response_payload = Payload(openFloor=outgoing_envelope) response = dict(response_payload) print(f"🦜 Sending response: {json.dumps(response, indent=2, default=str)}") return jsonify(response) except Exception as error: print(f"🦜 Error processing request: {error}") return jsonify({ 'error': 'Internal server error', 'message': str(error) }), 500
What's happening here:
- Validate that we received valid JSON
- Parse the payload into an Open Floor envelope
- Process it through our parrot agent
- Create and send the response as a properly formatted payload
Step 3: Creating the Entry Point
We end by creating a simple main.py
as our entry point:
from server import app import os if __name__ == '__main__': port = int(os.environ.get('PORT', 8080)) print(f"🦜 Parrot Agent server starting on port {port}") app.run(host='0.0.0.0', port=port, debug=True)
Step 4: Final Setup
Your project structure should now look like this:
parrot-agent/ ├── requirements.txt ├── parrot_agent.py ├── server.py └── main.py
Test Your Implementation
Run this to test:
python main.py
Send your manifest or utterance requests to http://localhost:8080/
to see if it's working! You can also download the simple single HTML file manifest and utterance chat azettl/openfloor-js-chat to test your agent locally.
If you found this guide useful, follow me for more and let me know what you build with it in the comments!
Top comments (1)
Nice read and cool setup, I will test it!