Use with AI Agents

curl -X POST "https://api.creativeforesight.io/v1/signup" \
  -H "Content-Type: application/json" \
  -d '{"email":"[email protected]","source":"agents guide"}'
 
export FORESIGHT_API_KEY=cf_live_...

import os
import urllib.error
import urllib.parse
import urllib.request
 
FORESIGHT_API = "https://api.creativeforesight.io"
 
 
def call_foresight(path: str, params: dict) -> tuple[str, bool]:
    """GET a Foresight API endpoint. Returns (json_body, is_error)."""
    headers = {}
    api_key = os.environ.get("FORESIGHT_API_KEY")  # cf_live_... from POST /v1/signup
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
    url = f"{FORESIGHT_API}{path}?{urllib.parse.urlencode(params)}"
    try:
        with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=30) as response:
            return response.read().decode(), False
    except urllib.error.HTTPError as error:
        return error.read().decode(), True
 
 
def search_indicators(query: str, limit: int = 5) -> tuple[str, bool]:
    return call_foresight("/v1/indicators", {"search": query, "limit": limit})
 
 
def get_latest_observation(indicator: str, region: str = "US") -> tuple[str, bool]:
    return call_foresight("/v1/observations/latest", {"indicator": indicator, "region": region})
 
 
TOOL_FUNCTIONS = {
    "search_indicators": search_indicators,
    "get_latest_observation": get_latest_observation,
}

SEARCH_INDICATORS_SCHEMA = {
    "type": "object",
    "properties": {
        "query": {"type": "string", "description": "Plain-language search, e.g. 'unemployment rate'."},
        "limit": {"type": "integer", "minimum": 1, "maximum": 20, "description": "Rows to return (default 5)."},
    },
    "required": ["query"],
}
 
GET_LATEST_OBSERVATION_SCHEMA = {
    "type": "object",
    "properties": {
        "indicator": {"type": "string", "description": "Indicator code from search_indicators, e.g. 'UNRATE'."},
        "region": {"type": "string", "description": "'US', a 2-digit state FIPS, or a 5-digit county FIPS (default 'US')."},
    },
    "required": ["indicator"],
}

import anthropic
 
from foresight_tools import GET_LATEST_OBSERVATION_SCHEMA, SEARCH_INDICATORS_SCHEMA, TOOL_FUNCTIONS
 
client = anthropic.Anthropic()
tools = [
    {
        "name": "search_indicators",
        "description": "Search the Foresight API indicator catalog. Returns indicator codes, units and access tiers.",
        "input_schema": SEARCH_INDICATORS_SCHEMA,
    },
    {
        "name": "get_latest_observation",
        "description": "Get the most recent value of a Foresight indicator for a region.",
        "input_schema": GET_LATEST_OBSERVATION_SCHEMA,
    },
]
messages = [{"role": "user", "content": "What is the latest US unemployment rate?"}]
 
response = client.messages.create(model="claude-opus-5", max_tokens=16000, tools=tools, messages=messages)
while response.stop_reason == "tool_use":
    messages.append({"role": "assistant", "content": response.content})
    results = []
    for block in response.content:
        if block.type == "tool_use":
            body, is_error = TOOL_FUNCTIONS[block.name](**block.input)
            results.append({"type": "tool_result", "tool_use_id": block.id, "content": body, "is_error": is_error})
    messages.append({"role": "user", "content": results})
    response = client.messages.create(model="claude-opus-5", max_tokens=16000, tools=tools, messages=messages)
 
print("".join(block.text for block in response.content if block.type == "text"))

import json
 
from openai import OpenAI
 
from foresight_tools import GET_LATEST_OBSERVATION_SCHEMA, SEARCH_INDICATORS_SCHEMA, TOOL_FUNCTIONS
 
client = OpenAI()
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_indicators",
            "description": "Search the Foresight API indicator catalog. Returns indicator codes, units and access tiers.",
            "parameters": SEARCH_INDICATORS_SCHEMA,
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_latest_observation",
            "description": "Get the most recent value of a Foresight indicator for a region.",
            "parameters": GET_LATEST_OBSERVATION_SCHEMA,
        },
    },
]
messages = [{"role": "user", "content": "What is the latest US unemployment rate?"}]
 
while True:
    message = client.chat.completions.create(model="gpt-5", messages=messages, tools=tools).choices[0].message
    if not message.tool_calls:
        break
    messages.append(message)
    for call in message.tool_calls:
        body, _ = TOOL_FUNCTIONS[call.function.name](**json.loads(call.function.arguments))
        messages.append({"role": "tool", "tool_call_id": call.id, "content": body})
 
print(message.content)

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
 
from foresight_tools import call_foresight
 
 
@tool
def search_indicators(query: str, limit: int = 5) -> str:
    """Search the Foresight API indicator catalog. Returns indicator codes, units and access tiers."""
    return call_foresight("/v1/indicators", {"search": query, "limit": limit})[0]
 
 
@tool
def get_latest_observation(indicator: str, region: str = "US") -> str:
    """Get the most recent value of a Foresight indicator. region is 'US', a 2-digit state FIPS, or a 5-digit county FIPS."""
    return call_foresight("/v1/observations/latest", {"indicator": indicator, "region": region})[0]
 
 
tools = {t.name: t for t in (search_indicators, get_latest_observation)}
llm = ChatAnthropic(model="claude-opus-5").bind_tools(list(tools.values()))
 
messages = [HumanMessage("What is the latest US unemployment rate?")]
while True:
    reply = llm.invoke(messages)
    messages.append(reply)
    if not reply.tool_calls:
        break
    for call in reply.tool_calls:
        messages.append(tools[call["name"]].invoke(call))  # a ToolMessage
 
print(reply.text)