OAuth Client Credentials Flow

import asyncio
from typing import Generator
 
from fastmcp import Client as FastMcpClient
import httpx
from kfinance.client.kfinance import Client as KfinanceClient
 
# Initialize a kfinance client.
# This client is responsible for refreshing the access token when it gets
# close to expiration after one hour.
# For a production environment, kensho-kfinance may be too heavyweight a
# dependency. In that case, you could just copy the refresh logic from
# https://github.com/kensho-technologies/kfinance/blob/3610e6423a99e25c22cdaa914da1e1e50e3552b3/kfinance/client/fetch.py#L141-L165
# or use a similar solution from requests or httpx
kfinance_client = KfinanceClient(client_id="<my_client_id>", private_key="<my_private_key>")
 
# The access token can be retrieved via the `access_token` attribute.
print(f"Access token: {kfinance_client.access_token[10:]}...")
 
# With the access token, you're now ready to make MCP requests.
# Fastmcp provides a decent non-agentic client: https://gofastmcp.com/clients/client
# Non-agentic means that the client doesn't translate queries like
# "Get me the price of MSFT." into specific tool calls. Instead, we
# have to specify the tool calls ourselves.
# There are agentic clients available but which library you want to use
# for this purpose probably depends on which model you intend to use.
 
# Before getting started with MCP requests, we define an httpx.Auth class,
# which injects the access token into each request.
 
class KfinanceAuth(httpx.Auth):
   """Httpx auth class that fetches access token from KfinanceClient."""
 
   def __init__(self, kfinance_client):
       self.kfinance_client = kfinance_client
 
   def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, httpx.Response, None]:
       """Inject the kfinance access token into the Authorization header."""
 
       # Fetch the current access token from the access_token attribute.
       access_token = self.kfinance_client.access_token
       request.headers["Authorization"] = f"Bearer {access_token}"
       yield request
 
fastmcp_client = FastMcpClient(
   transport="https://kfinance.kensho.com/integrations/mcp",
   auth=KfinanceAuth(kfinance_client=kfinance_client),
)
 
async def list_tools():
   """Fetch a list of all available tools."""
 
   async with fastmcp_client:
       tools = await fastmcp_client.list_tools()
       print(f"\nSuccessfully fetched a list of {len(tools)} tools, starting with {tools[0].name}.")
 
 
async def fetch_msft_2024_revenue():
   """Fetch Micorsoft's 2024 revenue.
 
   Usually, you won't have to generate all of the arguments yourself.
   Instead, an LLM will generate them for you, based on the tool information
   that it receives as part of the list_tools call.
   """
 
   async with fastmcp_client:
       result = await fastmcp_client.call_tool(
           "get_financial_line_item_from_identifiers",
           arguments={
               "identifiers": ["MSFT"],
               "line_item": "revenue",
               "start_year": 2024,
               "end_year": 2024,
           },
       )
       print(f"\n Microsoft's 2024 revenue is {result.data}")
 
 
asyncio.run(list_tools())
asyncio.run(fetch_msft_2024_revenue())