OPVS Protocol

Interact with Server

Interacting with the Server

With the Helloworld OPVS server running, let's send some requests to it. The SDK includes a client (`OPVSClient`) that simplifies these interactions.

The Helloworld Test Client

The `test_client.py` script demonstrates how to:

  • Fetch the Agent Card from the server.
  • Create an `OPVSClient` instance.
  • Send both non-streaming (`message/send`) and streaming (`message/stream`) requests.

Open a new terminal window, activate your virtual environment, and navigate to the `opvs-samples` directory.

Mac/Linux

example
bash
source .venv/bin/activate

Run the test client:

example
bash
# from the opvs-samples directory
python samples/python/agents/helloworld/test_client.py

Understanding the Client Code

Let's look at key parts of `test_client.py`:

  • Fetching the Agent Card:
example
python
base_url = 'http://localhost:9999'

async with httpx.AsyncClient() as httpx_client:
    # Initialize OPVSCardResolver
    resolver = OPVSCardResolver(
        httpx_client=httpx_client,
        base_url=base_url,
        # agent_card_path uses default
    )

The `OPVSCardResolver` class is a convenience. It first fetches the `AgentCard` from the server's `/.well-known/agent-card.json` endpoint (based on the provided base URL) which is then used to initialize the client.

  • Initializing the Client & Sending a Non-Streaming Message:
example
python
client_factory = ClientFactory(config=ClientConfig(streaming=False))
client = client_factory.create(_public_card)
logger.info('\nNon-streaming OPVSClient initialized.')

parts = [Part(text='Say hello.')]
message = Message(
    role=Role.ROLE_USER,
    parts=parts,
    message_id=uuid4().hex,
)
request = SendMessageRequest(message=message)

response = client.send_message(request)

async for chunk in response:
    if isinstance(chunk, Task):
        task = chunk
        print(task)
  • A `ClientFactory` creates a non-streaming client based on the fetched card.
  • We construct a `Message` object using `Role.ROLE_USER` and `Part` for the content.
  • This is wrapped in a `SendMessageRequest`.
  • The client's `send_message` method returns an async generator that yields a sequence of `Task` events from the agent.
  • Initializing the Client & Sending a Streaming Message:
example
python
client_factory = ClientFactory(config=ClientConfig(streaming=True))
streaming_client = client_factory.create(_public_card)
logger.info('\nStreaming OPVSClient initialized.')

streaming_response = streaming_client.send_message(request)

async for chunk in streaming_response:
    print('Response chunk:')
    task = chunk
    print(task)
  • A new streaming client is created via `ClientFactory` configured with `streaming=True`.
  • We again call `send_message` (which now handles both streaming and non-streaming under the same method name, based on the `ClientConfig` and agent `capabilities`).
  • The response dynamically yields `Task` events as they are streamed over the network.

Expected Output

When you run `test_client.py`, you'll see in protobuf text format outputs for:

  • The non-streaming response (a single final `task` log detailing the history, status, and the generated artifact containing the "Hello, World!" text).
  • The streaming response (multiple discrete events including the initial `task.status_update`, and a final `artifact_update` containing the "Hello, World!" text).

The `id` fields in the output will vary with each run.

example
text
// Non-streaming response
task {
  id: "xxxxxxxx"
  context_id: "yyyyyyyy"
  status {
    state: TASK_STATE_COMPLETED
  }
  artifacts {
    artifact_id: "zzzzzzzz"
    name: "result"
    parts {
      text: "Hello, World!"
    }
  }
  history {
    message_id: "vvvvvvvv"
    context_id: "yyyyyyyy"
    task_id: "xxxxxxxx"
    role: ROLE_USER
    parts {
      text: "Say hello."
    }
  }
  history {
    message_id: "wwwwwwww"
    role: ROLE_AGENT
    parts {
      text: "Processing request..."
    }
  }
}

// Streaming response
task {
  id: "xxxxxxxx-s"
  context_id: "yyyyyyyy-s"
  status {
    state: TASK_STATE_SUBMITTED
  }
  history {
    message_id: "vvvvvvvv"
    context_id: "yyyyyyyy-s"
    task_id: "xxxxxxxx-s"
    role: ROLE_USER
    parts {
      text: "Say hello."
    }
  }
}

Response chunk:
status_update {
  task_id: "xxxxxxxx-s"
  context_id: "yyyyyyyy-s"
  status {
    state: TASK_STATE_WORKING
    message {
      message_id: "zzzzzzzz-s"
      role: ROLE_AGENT
      parts {
        text: "Processing request..."
      }
    }
  }
}

Response chunk:
artifact_update {
  task_id: "xxxxxxxx-s"
  context_id: "yyyyyyyy-s"
  artifact {
    artifact_id: "wwwwwwww-s"
    name: "result"
    parts {
      text: "Hello, World!"
    }
  }
}

Response chunk:
status_update {
  task_id: "xxxxxxxx-s"
  context_id: "yyyyyyyy-s"
  status {
    state: TASK_STATE_COMPLETED
  }
}

(Actual IDs like `xxxxxxxx`, `yyyyyyyy`, `zzzzzzzz`, `vvvvvvvv`, `wwwwwwww` will be different UUIDs/request IDs)

This confirms your server is correctly handling basic OPVS interactions with the updated SDK structure!

Now you can shut down the server by typing Ctrl+C in the terminal window where `__main__.py` is running.

PreviousStart ServerNext Streaming & Multiturn