> ## Documentation Index
> Fetch the complete documentation index at: https://thethirdpenco-feat-tool-lifecycle-events.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
# RemoteAgent & connect()
> API reference for remote agent client proxy and connection function
The `RemoteAgent` class provides a client-side proxy that mimics the local `Agent` interface while communicating with remote agents via HTTP. The `connect()` function creates these proxies easily.
## Import
```python theme={null}
from ai_query.agents.remote import RemoteAgent, connect
from ai_query.agents.transport import HTTPTransport
```
## connect() Function
```python theme={null}
def connect(
url: str,
headers: dict[str, str] | None = None
) -> RemoteAgent
```
Create a connection to a remote agent, returning a `RemoteAgent` proxy.
### Parameters
Full URL to the agent endpoint. The last part of the path is extracted as the
agent\_id. Examples: - "[https://api.myapp.com/agents/researcher](https://api.myapp.com/agents/researcher)" -
"[https://lambda.execute-api.amazonaws.com/prod/worker](https://lambda.execute-api.amazonaws.com/prod/worker)"
Optional HTTP headers for authentication or other purposes.
### Returns
A RemoteAgent proxy that behaves like a local Agent instance.
### Examples
```python theme={null}
from ai_query.agents.remote import connect
# Basic connection
agent = connect("https://api.myapp.com/agents/chatbot")
response = await agent.chat("Hello!")
# With authentication
agent = connect(
"https://api.myapp.com/agents/researcher",
headers={"Authorization": "Bearer token123"}
)
# Different deployment targets
lambda_agent = connect("https://lambda.execute-api.amazonaws.com/prod/worker")
vercel_agent = connect("https://my-app.vercel.app/api/assistant")
fastapi_agent = connect("https://api.myservice.com/v1/agents/analyzer")
```
***
## RemoteAgent Class
### Constructor
```python theme={null}
def __init__(self, transport: HTTPTransport, agent_id: str) -> None
```
Direct constructor (usually you'll use `connect()` instead).
#### Parameters
HTTP transport instance for communication.
Agent ID to communicate with.
### Properties
#### id
```python theme={null}
@property
def id(self) -> str
```
Returns the agent ID.
```python theme={null}
agent = connect("https://api.myapp.com/agents/researcher")
print(agent.id) # "researcher"
```
***
## Methods
### chat
```python theme={null}
async def chat(
self,
message: str,
*,
signal: AbortSignal | None = None
) -> str
```
Send a chat message to the remote agent.
#### Parameters
Chat message to send.
Optional abort signal to cancel the request (not yet implemented).
#### Returns
The agent's text response.
#### Examples
```python theme={null}
agent = connect("https://api.myapp.com/agents/chatbot")
# Simple chat
response = await agent.chat("What's the weather like?")
print(response)
# Multi-turn conversation
response1 = await agent.chat("Tell me about Python")
response2 = await agent.chat("Can you give me an example?")
```
***
### stream
```python theme={null}
async def stream(
self,
message: str,
*,
signal: AbortSignal | None = None
) -> AsyncIterator[str]
```
Stream a response from the remote agent, yielding chunks as they're generated.
#### Parameters
Chat message to send.
Optional abort signal (not yet implemented).
#### Returns
Async iterator yielding response chunks.
#### Examples
```python theme={null}
agent = connect("https://api.myapp.com/agents/writer")
# Stream and print in real-time
print("Agent: ", end="", flush=True)
async for chunk in agent.stream("Write a short story about AI"):
print(chunk, end="", flush=True)
print() # New line
# Collect full response
full_response = ""
async for chunk in agent.stream("Summarize quantum computing"):
full_response += chunk
print(f"Full response: {full_response}")
```
***
### call
```python theme={null}
def call(self, *, agent_cls: type[T] | None = None) -> AgentCallProxy[T]
```
Returns a type-safe proxy for making fluent RPC calls to the remote agent.
#### Parameters
Optional agent class for type hints. If provided, enables IDE autocomplete for
that agent's methods.
#### Returns
A fluent proxy for method calls.
#### Examples
```python theme={null}
agent = connect("https://api.myapp.com/agents/data-analyzer")
# Untyped method calls
result = await agent.call().analyze_data(
data=[1, 2, 3, 4, 5],
method="mean"
)
# With type hints (assuming DataAnalyzer class exists)
from my_agents import DataAnalyzer
result = await agent.call(agent_cls=DataAnalyzer).analyze_data(
data=[1, 2, 3, 4, 5],
method="mean",
precision=2
)
# Chained calls
result = await agent.call().process().step1().step2()
```
***
### close
```python theme={null}
async def close(self) -> None
```
Close the underlying transport and clean up resources.
#### Examples
```python theme={null}
agent = connect("https://api.myapp.com/agents/worker")
try:
await agent.chat("Hello")
await agent.stream("Generate a story")
finally:
await agent.close() # Cleanup
```
***
## Complete Examples
### Basic Client Usage
```python theme={null}
from ai_query.agents.remote import connect
async def chat_with_agents():
# Connect to different specialized agents
researcher = connect("https://api.myapp.com/agents/researcher")
writer = connect("https://api.myapp.com/agents/writer")
summarizer = connect(
"https://lambda.execute-api.amazonaws.com/prod/summarizer",
headers={"X-API-Key": "aws-key"}
)
try:
# Research phase
research = await researcher.chat("What are the latest trends in AI?")
print(f"Research: {research}")
# Writing phase
article = await writer.chat(f"Write an article based on: {research}")
print(f"Article: {article[:100]}...")
# Summary phase (streaming)
print("Summary: ", end="", flush=True)
async for chunk in summarizer.stream(f"Summarize: {article}"):
print(chunk, end="", flush=True)
print()
finally:
# Cleanup connections
await researcher.close()
await writer.close()
await summarizer.close()
```
### RPC Method Calls
```python theme={null}
from ai_query.agents.remote import connect
from my_agents import DataProcessor, ImageAnalyzer
async def rpc_examples():
# Data processing agent
data_agent = connect("https://api.myapp.com/agents/data-processor")
# Type-safe calls
result = await data_agent.call(agent_cls=DataProcessor).analyze_dataset(
data=[1, 2, 3, 4, 5],
method="statistical",
confidence_level=0.95
)
print(f"Analysis result: {result}")
# Image analysis agent
image_agent = connect("https://api.myapp.com/agents/image-analyzer")
analysis = await image_agent.call(agent_cls=ImageAnalyzer).detect_objects(
image_url="https://example.com/image.jpg",
confidence_threshold=0.8
)
print(f"Objects detected: {analysis['objects']}")
```
### Error Handling and Timeouts
```python theme={null}
import asyncio
from ai_query.agents.remote import connect
async def robust_agent_calls():
agent = connect("https://api.myapp.com/agents/unreliable")
try:
# Add timeout
response = await asyncio.wait_for(
agent.chat("Complex task"),
timeout=10.0
)
print(response)
except asyncio.TimeoutError:
print("Agent took too long to respond")
except Exception as e:
print(f"Agent communication failed: {e}")
finally:
await agent.close()
async def retry_on_failure():
agent = connect("https://api.myapp.com/agents/flaky")
max_retries = 3
for attempt in range(max_retries):
try:
response = await agent.chat("Important task")
return response
except Exception as e:
if attempt == max_retries - 1:
raise
print(f"Attempt {attempt + 1} failed, retrying...")
await asyncio.sleep(1) # Backoff
await agent.close()
```
### Serverless Integration Examples
```python theme={null}
# AWS Lambda
lambda_agent = connect(
"https://lambda123.execute-api.us-east-1.amazonaws.com/prod/text-processor",
headers={"X-API-Key": "your-lambda-key"}
)
# Vercel
vercel_agent = connect("https://my-app.vercel.app/api/agent")
# FastAPI service
fastapi_agent = connect(
"https://api.myservice.com/v1/agents/chatbot",
headers={"Authorization": "Bearer your-jwt-token"}
)
# Railway/Render
railway_agent = connect("https://my-app.up.railway.app/agent")
async def multi_platform_workflow():
# Use different serverless agents interchangeably
text = await lambda_agent.chat("Process this text")
analysis = await vercel_agent.call().analyze(text)
final_result = await fastapi_agent.chat(f"Finalize: {analysis}")
return final_result
```
### Streaming Example with Progress
```python theme={null}
from ai_query.agents.remote import connect
import time
async def streaming_with_progress():
agent = connect("https://api.myapp.com/agents/long-writer")
message = "Write a comprehensive guide to machine learning"
chunk_count = 0
start_time = time.time()
print("Generating response...")
async for chunk in agent.stream(message):
chunk_count += 1
print(chunk, end="", flush=True)
# Optional: Show progress every 10 chunks
if chunk_count % 10 == 0:
elapsed = time.time() - start_time
print(f"\n[Progress: {chunk_count} chunks, {elapsed:.1f}s]", end="", flush=True)
total_time = time.time() - start_time
print(f"\n\nDone! {chunk_count} chunks in {total_time:.1f} seconds")
await agent.close()
```