Skip to main content

​
Class: Http

class Http:
    def fetch(
        self,
        url: str,
        *,
        method: str = "GET",
        headers: dict[str, str] | None = None,
        body: str | None = None,
        timeout_ms: int | None = None,
    ) -> HttpResponse: ...

​
Methods

​
fetch()

Make an HTTP request through the host. Parameters:
ParameterTypeRequiredDefaultDescription
urlstrYes-Target URL
methodstrNo"GET"HTTP method (GET, POST, PUT, PATCH, DELETE, HEAD)
headersdict[str, str] | NoneNoNoneRequest headers
bodystr | NoneNoNoneRequest body (string)
timeout_msint | NoneNoNoneRequest timeout in milliseconds
Returns: HttpResponse Raises: HttpError on network-level failure (DNS, TLS, timeout, connection)

​
HttpResponse

@dataclass
class HttpResponse:
    status: int              # HTTP status code
    headers: dict[str, str] # Response headers
    body: str               # Response body as string

    def json(self) -> Any:   # Parse body as JSON
        return json.loads(self.body)

​
Basic GET

response = ctx.http.fetch("https://api.example.com/data")

if response.status == 200:
    data = response.json()
    return ok({"data": data})
elif response.status == 404:
    return ok({"found": False})
else:
    return err(f"API error: {response.status}")

​
POST with JSON

import json

response = ctx.http.fetch(
    "https://api.example.com/items",
    method="POST",
    headers={
        "Content-Type": "application/json",
        "Authorization": f"Bearer {ctx.env['API_KEY']}",
    },
    body=json.dumps({"name": "New Item", "value": 42}),
)

result = response.json()
return ok({"created_id": result["id"]})

​
Error Handling

from friday_agent_sdk import HttpError, agent, err, ok

@agent(id="resilient", version="1.0.0", description="Handles HTTP failures")
def execute(prompt, ctx):
    try:
        response = ctx.http.fetch("https://api.example.com/data")
    except HttpError as e:
        # Network failure: DNS, TLS, connection, timeout
        return err(f"Request failed: {e}")

    # HTTP error status (4xx, 5xx) is NOT an exception
    if response.status >= 500:
        return err(f"Server error: {response.status}")

    if response.status == 429:
        return err("Rate limited. Please retry later.")

    if response.status >= 400:
        return err(f"Client error: {response.status} - {response.body[:200]}")

    return ok({"data": response.json()})

​
Timeout

response = ctx.http.fetch(
    "https://slow-api.example.com/data",
    timeout_ms=30000,  # 30 seconds
)

​
Methods

All HTTP methods are supported: 
ctx.http.fetch(url, method="GET")      # Default
ctx.http.fetch(url, method="POST", body="...")
ctx.http.fetch(url, method="PUT", body="...")
ctx.http.fetch(url, method="PATCH", body="...")
ctx.http.fetch(url, method="DELETE")
ctx.http.fetch(url, method="HEAD")

​
Headers

Case-insensitive header dict: 
response = ctx.http.fetch(
    url,
    headers={
        "Accept": "application/json",
        "User-Agent": "my-agent/1.0",
        "X-Custom-Header": "value",
    },
)

# Access response headers
content_type = response.headers.get("content-type", "")
rate_limit = response.headers.get("x-ratelimit-remaining")

​
Creating Artifacts

Common pattern: create Friday platform artifacts via HTTP API: 
import json

response = ctx.http.fetch(
    f"{ctx.config.get('platformUrl', 'http://localhost:18080')}/api/artifacts",
    method="POST",
    headers={"Content-Type": "application/json"},
    body=json.dumps({
        "data": {
            "type": "analysis",
            "version": 1,
            "data": analysis_result,
        },
        "title": "Analysis Report",
        "summary": "Comprehensive analysis",
    }),
)

if response.status < 400:
    artifact = response.json().get("artifact", {})
    artifact_id = artifact.get("id")
    ...

​
REST API Patterns

base_url = "https://api.service.com/v1"

# GET collection
response = ctx.http.fetch(f"{base_url}/items")
items = response.json()["items"]

# GET item
response = ctx.http.fetch(f"{base_url}/items/{item_id}")
item = response.json()

# POST create
response = ctx.http.fetch(
    f"{base_url}/items",
    method="POST",
    headers={"Content-Type": "application/json"},
    body=json.dumps({"name": "New"}),
)
new_item = response.json()

# PUT update
response = ctx.http.fetch(
    f"{base_url}/items/{item_id}",
    method="PUT",
    headers={"Content-Type": "application/json"},
    body=json.dumps({"name": "Updated"}),
)

# DELETE
response = ctx.http.fetch(
    f"{base_url}/items/{item_id}",
    method="DELETE",
)

​
Authentication

# Bearer token
response = ctx.http.fetch(
    url,
    headers={"Authorization": f"Bearer {ctx.env['TOKEN']}"},
)

# Basic auth (construct manually)
import base64
credentials = base64.b64encode(b"user:pass").decode()
response = ctx.http.fetch(
    url,
    headers={"Authorization": f"Basic {credentials}"},
)

# API key in header
response = ctx.http.fetch(
    url,
    headers={"X-API-Key": ctx.env['API_KEY']},
)

​
Query Parameters

Construct URL with parameters: 
import urllib.parse

params = {"q": "search query", "limit": 10}
query = urllib.parse.urlencode(params)
url = f"https://api.example.com/search?{query}"

response = ctx.http.fetch(url)

​
Response Body Limits

  • 5MB limit enforced by platform
  • Exceeding returns truncated or error response
  • For large payloads, consider streaming (not yet available)

​
URL Allowlists

Not yet implemented. Currently all outbound HTTPS requests are allowed.

​
Why Not Use requests or httpx?

Agents run as native Python processes. You can pip install requests or httpx if needed. Host-provided HTTP is still preferred for audit logging, rate limiting, TLS management, and response limits.

​
See Also

How to Make HTTP Requests

Task-oriented guide