Build with AWP

In this tutorial, you’ll learn how to build an HTTP endpoint that is compatible with the AWP protocol.

Prerequisites

Make sure to have Python and Poetry installed.

Setup a New Project with Poetry

First, let’s create a new project and set up Poetry for dependency management:

poetry new my-endpoint && cd my-endpoint

Install Dependencies

Now, let’s install the necessary packages:

poetry add agentwire openai fastapi uvicorn

Create a Basic Endpoint with FastAPI

Create a new file called my_endpoint/main.py with the following code:

from fastapi import FastAPI, Request
import json
from agentwire.core.types import RunAgentInput

app = FastAPI(title="AWP Endpoint")

@app.post("/awp")
async def my_endpoint():
    return { "message": "Hello World" }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run and Test Your Endpoint

Start the server with:

poetry run uvicorn my_endpoint.main:app --reload

In another terminal, test your endpoint is running using curl:

curl -X POST http://localhost:8000/awp

You should see the following response:

{ "message": "Hello World" }

Parsing AWP Input

Next let’s update our endpoint to properly parse the incoming AWP request using the RunAgentInput Pydantic model:

from fastapi import FastAPI, Request, HTTPException
from agentwire.core import RunAgentInput, Message

app = FastAPI(title="AWP Endpoint")

@app.post("/awp")
async def my_endpoint(input_data: RunAgentInput):
    thread_id = input_data.thread_id

    return { "message": "Hello World from " + thread_id }

FastAPI automatically validates the incoming request against the RunAgentInput schema. If the request doesn’t match the expected format, it will return a 422 Validation Error with details about what went wrong.

Add Event Streaming

AWP supports streaming events using Server-Sent Events (SSE). Let’s modify our /awp endpoint to stream events back to the client:

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
from agentwire.core import RunAgentInput, Message, EventType, RunStartedEvent, RunFinishedEvent
from agentwire.encoder import EventEncoder

app = FastAPI(title="AWP Endpoint")

@app.post("/awp")
async def my_endpoint(input_data: RunAgentInput):
    async def event_generator():
        # Create an event encoder to properly format SSE events
        encoder = EventEncoder()

        # Send run started event
        yield encoder.encode(
          RunStartedEvent(
            type=EventType.RUN_STARTED,
            thread_id=input_data.thread_id,
            run_id=input_data.run_id
          )
        )

        # Send run finished event
        yield encoder.encode(
          RunFinishedEvent(
            type=EventType.RUN_FINISHED,
            thread_id=input_data.thread_id,
            run_id=input_data.run_id
          )
        )

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Awesome! We are already sending RunStartedEvent and RunFinishedEvent events, which gives us a basic AWP compliant endpoint. Now let’s make it do something useful.

Implementing Basic Chat

Let’s enhance our endpoint to call OpenAI’s API and stream the responses back as AWP events:

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from agentwire.core import (
  RunAgentInput,
  Message,
  EventType,
  RunStartedEvent,
  RunFinishedEvent,
  TextMessageStartEvent,
  TextMessageContentEvent,
  TextMessageEndEvent
)
from agentwire.encoder import EventEncoder
import uuid
from openai import OpenAI

app = FastAPI(title="AWP Endpoint")

@app.post("/awp")
async def my_endpoint(input_data: RunAgentInput):
    async def event_generator():
        # Create an event encoder to properly format SSE events
        encoder = EventEncoder()

        # Send run started event
        yield encoder.encode(
          RunStartedEvent(
            type=EventType.RUN_STARTED,
            thread_id=input_data.thread_id,
            run_id=input_data.run_id
          )
        )

        # Initialize OpenAI client
        client = OpenAI()

        # Generate a message ID for the assistant's response
        message_id = uuid.uuid4()

        # Send text message start event
        yield encoder.encode(
            TextMessageStartEvent(
                type=EventType.TEXT_MESSAGE_START,
                message_id=message_id,
                role="assistant"
            )
        )

        # Create a streaming completion request
        stream = client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=openai_messages,
            stream=True
        )

        # Process the streaming response and send content events
        for chunk in stream:
            if hasattr(chunk.choices[0].delta, "content") and chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                yield encoder.encode(
                    TextMessageContentEvent(
                        type=EventType.TEXT_MESSAGE_CONTENT,
                        message_id=message_id,
                        delta=content
                    )
                )

        # Send text message end event
        yield encoder.encode(
            TextMessageEndEvent(
                type=EventType.TEXT_MESSAGE_END,
                message_id=message_id
            )
        )

        # Send run finished event
        yield encoder.encode(
          RunFinishedEvent(
            type=EventType.RUN_FINISHED,
            thread_id=input_data.thread_id,
            run_id=input_data.run_id
          )
        )

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

You’ll need to set your OpenAI API key as an environment variable and then restart the server:

export OPENAI_API_KEY=your-api-key
poetry run uvicorn my_endpoint.main:app --reload

This implementation creates a fully functional AWP endpoint that processes messages and streams back the responses in real-time.

Prerequisites

Make sure to have Python and Poetry installed.

Setup a New Project with Poetry

First, let’s create a new project and set up Poetry for dependency management:

poetry new my-endpoint && cd my-endpoint

Install Dependencies

Now, let’s install the necessary packages:

poetry add agentwire openai fastapi uvicorn

Create a Basic Endpoint with FastAPI

Create a new file called my_endpoint/main.py with the following code:

from fastapi import FastAPI, Request
import json
from agentwire.core.types import RunAgentInput

app = FastAPI(title="AWP Endpoint")

@app.post("/awp")
async def my_endpoint():
    return { "message": "Hello World" }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run and Test Your Endpoint

Start the server with:

poetry run uvicorn my_endpoint.main:app --reload

In another terminal, test your endpoint is running using curl:

curl -X POST http://localhost:8000/awp

You should see the following response:

{ "message": "Hello World" }

Parsing AWP Input

Next let’s update our endpoint to properly parse the incoming AWP request using the RunAgentInput Pydantic model:

from fastapi import FastAPI, Request, HTTPException
from agentwire.core import RunAgentInput, Message

app = FastAPI(title="AWP Endpoint")

@app.post("/awp")
async def my_endpoint(input_data: RunAgentInput):
    thread_id = input_data.thread_id

    return { "message": "Hello World from " + thread_id }

Add Event Streaming

AWP supports streaming events using Server-Sent Events (SSE). Let’s modify our /awp endpoint to stream events back to the client:

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
from agentwire.core import RunAgentInput, Message, EventType, RunStartedEvent, RunFinishedEvent
from agentwire.encoder import EventEncoder

app = FastAPI(title="AWP Endpoint")

@app.post("/awp")
async def my_endpoint(input_data: RunAgentInput):
    async def event_generator():
        # Create an event encoder to properly format SSE events
        encoder = EventEncoder()

        # Send run started event
        yield encoder.encode(
          RunStartedEvent(
            type=EventType.RUN_STARTED,
            thread_id=input_data.thread_id,
            run_id=input_data.run_id
          )
        )

        # Send run finished event
        yield encoder.encode(
          RunFinishedEvent(
            type=EventType.RUN_FINISHED,
            thread_id=input_data.thread_id,
            run_id=input_data.run_id
          )
        )

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Awesome! We are already sending RunStartedEvent and RunFinishedEvent events, which gives us a basic AWP compliant endpoint. Now let’s make it do something useful.

Implementing Basic Chat

Let’s enhance our endpoint to call OpenAI’s API and stream the responses back as AWP events:

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from agentwire.core import (
  RunAgentInput,
  Message,
  EventType,
  RunStartedEvent,
  RunFinishedEvent,
  TextMessageStartEvent,
  TextMessageContentEvent,
  TextMessageEndEvent
)
from agentwire.encoder import EventEncoder
import uuid
from openai import OpenAI

app = FastAPI(title="AWP Endpoint")

@app.post("/awp")
async def my_endpoint(input_data: RunAgentInput):
    async def event_generator():
        # Create an event encoder to properly format SSE events
        encoder = EventEncoder()

        # Send run started event
        yield encoder.encode(
          RunStartedEvent(
            type=EventType.RUN_STARTED,
            thread_id=input_data.thread_id,
            run_id=input_data.run_id
          )
        )

        # Initialize OpenAI client
        client = OpenAI()

        # Generate a message ID for the assistant's response
        message_id = uuid.uuid4()

        # Send text message start event
        yield encoder.encode(
            TextMessageStartEvent(
                type=EventType.TEXT_MESSAGE_START,
                message_id=message_id,
                role="assistant"
            )
        )

        # Create a streaming completion request
        stream = client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=openai_messages,
            stream=True
        )

        # Process the streaming response and send content events
        for chunk in stream:
            if hasattr(chunk.choices[0].delta, "content") and chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                yield encoder.encode(
                    TextMessageContentEvent(
                        type=EventType.TEXT_MESSAGE_CONTENT,
                        message_id=message_id,
                        delta=content
                    )
                )

        # Send text message end event
        yield encoder.encode(
            TextMessageEndEvent(
                type=EventType.TEXT_MESSAGE_END,
                message_id=message_id
            )
        )

        # Send run finished event
        yield encoder.encode(
          RunFinishedEvent(
            type=EventType.RUN_FINISHED,
            thread_id=input_data.thread_id,
            run_id=input_data.run_id
          )
        )

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

You’ll need to set your OpenAI API key as an environment variable and then restart the server:

export OPENAI_API_KEY=your-api-key
poetry run uvicorn my_endpoint.main:app --reload

This implementation creates a fully functional AWP endpoint that processes messages and streams back the responses in real-time.

Prerequisites

Make sure to have Node.js (v16 or later) and npm or yarn installed.

Setup a New Project

First, let’s create a new project and set up npm with TypeScript:

mkdir awp-endpoint && cd awp-endpoint
npm init -y
npm install typescript ts-node @types/node @types/express --save-dev
npx tsc --init

Install Dependencies

Install the necessary packages:

npm install express openai @agentwire/core @agentwire/encoder uuid
npm install @types/uuid --save-dev

Create a Basic Endpoint with Express

Create a new file called src/server.ts with the following code:

import express from "express"
import { Request, Response } from "express"

const app = express()

app.use(express.json())

app.post("/awp", (req: Request, res: Response) => {
  res.json({ message: "Hello World" })
})

app.listen(8000, () => {
  console.log("Server running on http://localhost:8000")
})

Run and Test Your Endpoint

Start the server with:

npx ts-node src/server.ts

In another terminal, test your endpoint is running using curl:

curl -X POST http://localhost:8000/awp

You should see the following response:

{ "message": "Hello World" }

Parsing AWP Input

Next let’s update our endpoint to properly parse the incoming AWP request using the RunAgentInput schema:

import express, { Request, Response } from "express"
import { RunAgentInputSchema, RunAgentInput } from "@agentwire/core"

const app = express()

app.use(express.json())

app.post("/awp", (req: Request, res: Response) => {
  try {
    // Parse and validate the request body
    const input: RunAgentInput = RunAgentInputSchema.parse(req.body)
    res.json({ message: `Hello World from ${input.threadId}` })
  } catch (error) {
    res.status(422).json({ error: (error as Error).message })
  }
})

app.listen(8000, () => {
  console.log("Server running on http://localhost:8000")
})

Express with zod validation ensures the incoming request conforms to the AWP protocol format. If the request doesn’t match the expected format, it will return a 422 Validation Error with details about what went wrong.

Add Event Streaming

AWP supports streaming events using Server-Sent Events (SSE). Let’s modify our /awp endpoint to stream events back to the client:

import express, { Request, Response } from "express"
import { RunAgentInputSchema, RunAgentInput, EventType } from "@agentwire/core"
import { EventEncoder } from "@agentwire/encoder"

const app = express()

app.use(express.json())

app.post("/awp", async (req: Request, res: Response) => {
  try {
    // Parse and validate the request body
    const input: RunAgentInput = RunAgentInputSchema.parse(req.body)

    // Set up SSE headers
    res.setHeader("Content-Type", "text/event-stream")
    res.setHeader("Cache-Control", "no-cache")
    res.setHeader("Connection", "keep-alive")

    // Create an event encoder
    const encoder = new EventEncoder()

    // Send run started event
    const runStarted = {
      type: EventType.RUN_STARTED,
      threadId: input.threadId,
      runId: input.runId,
    }
    res.write(encoder.encode(runStarted))

    // Send run finished event
    const runFinished = {
      type: EventType.RUN_FINISHED,
      threadId: input.threadId,
      runId: input.runId,
    }
    res.write(encoder.encode(runFinished))

    // End the response
    res.end()
  } catch (error) {
    res.status(422).json({ error: (error as Error).message })
  }
})

app.listen(8000, () => {
  console.log("Server running on http://localhost:8000")
})

Implementing Basic Chat

Let’s enhance our endpoint to call OpenAI’s API and stream the responses back as AWP events:

import express, { Request, Response } from "express"
import {
  RunAgentInputSchema,
  RunAgentInput,
  EventType,
  Message,
} from "@agentwire/core"
import { EventEncoder } from "@agentwire/encoder"
import { OpenAI } from "openai"
import { v4 as uuidv4 } from "uuid"

const app = express()

app.use(express.json())

app.post("/awp", async (req: Request, res: Response) => {
  try {
    // Parse and validate the request body
    const input: RunAgentInput = RunAgentInputSchema.parse(req.body)

    // Set up SSE headers
    res.setHeader("Content-Type", "text/event-stream")
    res.setHeader("Cache-Control", "no-cache")
    res.setHeader("Connection", "keep-alive")

    // Create an event encoder
    const encoder = new EventEncoder()

    // Send run started event
    const runStarted = {
      type: EventType.RUN_STARTED,
      threadId: input.threadId,
      runId: input.runId,
    }
    res.write(encoder.encode(runStarted))

    // Initialize OpenAI client
    const client = new OpenAI()

    // Convert AWP messages to OpenAI messages format
    const openaiMessages = input.messages
      .filter((msg: Message) =>
        ["user", "system", "assistant"].includes(msg.role),
      )
      .map((msg: Message) => ({
        role: msg.role as "user" | "system" | "assistant",
        content: msg.content || "",
      }))

    // Generate a message ID for the assistant's response
    const messageId = uuidv4()

    // Send text message start event
    const textMessageStart = {
      type: EventType.TEXT_MESSAGE_START,
      messageId,
      role: "assistant",
    }
    res.write(encoder.encode(textMessageStart))

    // Create a streaming completion request
    const stream = await client.chat.completions.create({
      model: "gpt-3.5-turbo",
      messages: openaiMessages,
      stream: true,
    })

    // Process the streaming response and send content events
    for await (const chunk of stream) {
      if (chunk.choices[0]?.delta?.content) {
        const content = chunk.choices[0].delta.content
        const textMessageContent = {
          type: EventType.TEXT_MESSAGE_CONTENT,
          messageId,
          delta: content,
        }
        res.write(encoder.encode(textMessageContent))
      }
    }

    // Send text message end event
    const textMessageEnd = {
      type: EventType.TEXT_MESSAGE_END,
      messageId,
    }
    res.write(encoder.encode(textMessageEnd))

    // Send run finished event
    const runFinished = {
      type: EventType.RUN_FINISHED,
      threadId: input.threadId,
      runId: input.runId,
    }
    res.write(encoder.encode(runFinished))

    // End the response
    res.end()
  } catch (error) {
    res.status(422).json({ error: (error as Error).message })
  }
})

app.listen(8000, () => {
  console.log("Server running on http://localhost:8000")
})

You’ll need to set your OpenAI API key as an environment variable and then start the server:

export OPENAI_API_KEY=your-api-key
npx ts-node src/server.ts

This implementation creates a fully functional AWP endpoint that processes messages and streams back the responses in real-time.

Get Started

Concepts

Tutorials

Development

Prerequisites

Setup a New Project with Poetry

Install Dependencies

Create a Basic Endpoint with FastAPI

Run and Test Your Endpoint

Parsing AWP Input

Add Event Streaming

Implementing Basic Chat

Prerequisites

Setup a New Project with Poetry

Install Dependencies

Create a Basic Endpoint with FastAPI

Run and Test Your Endpoint

Parsing AWP Input

Add Event Streaming

Implementing Basic Chat

Prerequisites

Setup a New Project

Install Dependencies

Create a Basic Endpoint with Express

Run and Test Your Endpoint

Parsing AWP Input

Add Event Streaming

Implementing Basic Chat

Get Started

Concepts

Tutorials

Development

​Prerequisites

​Setup a New Project with Poetry

​Install Dependencies

​Create a Basic Endpoint with FastAPI

​Run and Test Your Endpoint

​Parsing AWP Input

​Add Event Streaming

​Implementing Basic Chat

​Prerequisites

​Setup a New Project with Poetry

​Install Dependencies

​Create a Basic Endpoint with FastAPI

​Run and Test Your Endpoint

​Parsing AWP Input

​Add Event Streaming

​Implementing Basic Chat

​Prerequisites

​Setup a New Project

​Install Dependencies

​Create a Basic Endpoint with Express

​Run and Test Your Endpoint

​Parsing AWP Input

​Add Event Streaming

​Implementing Basic Chat

Prerequisites

Setup a New Project with Poetry

Install Dependencies

Create a Basic Endpoint with FastAPI

Run and Test Your Endpoint

Parsing AWP Input

Add Event Streaming

Implementing Basic Chat

Prerequisites

Setup a New Project with Poetry

Install Dependencies

Create a Basic Endpoint with FastAPI

Run and Test Your Endpoint

Parsing AWP Input

Add Event Streaming

Implementing Basic Chat

Prerequisites

Setup a New Project

Install Dependencies

Create a Basic Endpoint with Express

Run and Test Your Endpoint

Parsing AWP Input

Add Event Streaming

Implementing Basic Chat