graphplug

Microsoft Graph for Python, one call at a time. Send a mail, book a Teams meeting, upload a file or look someone up in the directory without writing Graph's JSON by hand. Sign-in, retries, throttling, paging and batching are handled for you.

pip install graphplug

Python 3.10+ · async · depends only on Microsoft's azure-identity and msgraph-core

Contents

1. Quick start
2. Signing in
3. Permissions (Scopes)
4. Mail
5. Calendar and meetings
6. Files (OneDrive)
7. Teams and chats
8. People and the directory
9. Methods every resource has
10. Any other Graph endpoint
11. Paging
12. Batching
13. Downloads and uploads
14. Errors
15. Logging
16. Client lifetime and concurrency

1. Quick start

Everything is async, so a script wraps its work in async def main() and runs it with asyncio.run.

import asyncio
from graphplug import GraphClient, Scopes

TENANT = "your-tenant-id"
CLIENT = "your-app-client-id"

async def main():
    # Prints a code; you sign in on any browser, then the script continues.
    graph = await GraphClient.device_code(TENANT, CLIENT, Scopes.MAIL_SEND)

    async with graph:
        await graph.mail.send(
            to="alice@example.com",
            subject="Quarterly report",
            body="<p>Attached.</p>", html=True,
            attachments=["report.pdf"],
        )

asyncio.run(main())

The client exposes five resources plus generic access to all of Graph:

2. Signing in

Pick an access model first

Mail.Read as an application permission reads every mailbox in the tenant. As a delegated permission it reads only the signed-in person's mail. Choose deliberately.

Ways to build a client

Every constructor also accepts max_concurrency (default 12), and the Entra ones accept authority_host for national clouds.

Application access, from environment variables

async with GraphClient.from_env() as graph:
    async for user in graph.paged("/users", select="id,displayName,mail"):
        print(user["displayName"], user["mail"])

Application access, explicit

graph = GraphClient.app_only(TENANT, CLIENT, client_secret=SECRET)

Delegated, device code

scopes = Scopes.combine(Scopes.MAIL_SEND, Scopes.USER_READ)
graph = await GraphClient.device_code(TENANT, CLIENT, scopes)

Delegated, device code with your own prompt

pending = await GraphClient.begin_device_code(TENANT, CLIENT, Scopes.CALENDARS_READ_WRITE)
print(pending.message)            # or pending.user_code and pending.verification_uri
graph = await pending.complete()  # waits until the person has signed in

Delegated, browser on this machine

graph = await GraphClient.interactive(TENANT, CLIENT, Scopes.FILES_READ_WRITE,
                                      redirect_uri="http://localhost:8400")

The redirect_uri must match the app registration exactly: http://localhost:8400 and http://localhost:8400/ are different to Entra.

Any azure-identity credential

from azure.identity.aio import ManagedIdentityCredential

graph = GraphClient.from_credential(ManagedIdentityCredential())

Sync credentials work too and run on a worker thread. A credential you pass in stays yours: closing the client does not close it.

Acting for a mailbox with application access

An app-only client has no signed-in person, so there is no "me". Every mail, calendar and files method, and the shared methods in §9, take a user argument: the mailbox to act as, given as an email address, user principal name or object id. Leave it out and the call uses the signed-in person, as before.

from datetime import datetime, timedelta, timezone

MY_MAILBOX = "you@yourcompany.com"   # your work mailbox: the sender and the meeting organiser

async with GraphClient.from_env() as graph:
    # A mail from your mailbox, with no one signed in
    await graph.mail.send(
        to="team@yourcompany.com", subject="Nightly report", body="All green.",
        user=MY_MAILBOX,
    )

    # A meeting organised from your calendar. Invitations go out from your mailbox.
    start = datetime(2026, 10, 1, 9, 0, tzinfo=timezone.utc)
    await graph.calendar.schedule(
        subject="Design review",
        start=start, end=start + timedelta(hours=1),
        attendees=["alice@yourcompany.com", "bob@yourcompany.com"],  # required
        optional_attendees=["carol@yourcompany.com"],                # optional
        online=True,                                                 # adds a Teams link
        user=MY_MAILBOX,
    )

    # Block two hours of focus time in your own calendar (no attendees; shows as busy)
    await graph.calendar.schedule(
        subject="Focus time",
        start=start + timedelta(hours=2), end=start + timedelta(hours=4),
        user=MY_MAILBOX,
    )

    # Are the attendees free? Asked from your mailbox
    busy = await graph.calendar.free_busy(
        ["alice@yourcompany.com", "bob@yourcompany.com"],
        start, start + timedelta(hours=8),
        user=MY_MAILBOX,
    )

Leave both attendee lists out and the event is a private block in user's calendar.

Set-up in Entra. Add application permissions to the app registration and grant admin consent: Mail.Send to send, Mail.ReadWrite to read and tidy mail, Calendars.ReadWrite for calendars, Files.ReadWrite.All for OneDrive. These reach every mailbox in the tenant, so in Exchange Online restrict the app to the mailboxes it needs with RBAC for Applications (or an application access policy).

Not available to apps. Graph does not let an app post Teams channel or chat messages as a person (outside data migration), so teams.post, teams.reply and teams.send_chat still need a signed-in person.

3. Permissions (Scopes)

Delegated sign-ins need explicit scopes; there is no default. Scopes names them so you do not have to remember Graph's spelling.

Combine groups with Scopes.combine(...), which keeps order and drops duplicates:

scopes = Scopes.combine(Scopes.MAIL_SEND, Scopes.CALENDARS_READ_WRITE, Scopes.USER_READ)

Application access ignores these: it always uses the permissions an admin granted to the app.

4. Mail — graph.mail

Works on the signed-in person's mailbox, or on any mailbox when you pass user= (every method takes it; required under application access).

send parameters

await graph.mail.send(
    to=["alice@example.com", "bob@example.com"],
    cc="team@example.com",
    subject="Build finished",
    body="<b>All green.</b>", html=True,
    attachments=["results.csv"],
)

# Read unread mail from the last day, then mark it read
from datetime import datetime, timedelta, timezone
since = datetime.now(timezone.utc) - timedelta(days=1)
async for message in graph.mail.inbox(unread_only=True, since=since):
    print(message["subject"], "from", message["from"]["emailAddress"]["address"])
    await graph.mail.mark_read(message["id"])

# Search (Graph does not allow search together with filters or sorting)
async for message in graph.mail.inbox(search="invoice"):
    print(message["subject"])

# Reply and forward
await graph.mail.reply(message_id, "Thanks, on it.", reply_all=True)
await graph.mail.forward(message_id, to="carol@example.com", comment="FYI")

# Send many at once; each entry takes the same arguments as send()
results = await graph.mail.send_many([
    {"to": "a@example.com", "subject": "Hi A", "body": "..."},
    {"to": "b@example.com", "subject": "Hi B", "body": "..."},
])
failed = [r for r in results if r["status"] >= 400]

For attachments over about 3 MB, upload the file with graph.files.upload and send a share_link instead.

5. Calendar and meetings — graph.calendar

Works on the signed-in person's calendar, or on anyone's (a person or a room) when you pass user= (every method takes it; required under application access). New events show as busy, which is what blocks the time.

schedule parameters

from datetime import datetime, timedelta, timezone

start = datetime(2026, 10, 1, 14, 0, tzinfo=timezone.utc)
event = await graph.calendar.schedule(
    subject="Design review",
    start=start, end=start + timedelta(minutes=45),
    attendees=["alice@example.com", "bob@example.com"],
    online=True,
    reminder_minutes=10,
)
print(event["onlineMeeting"]["joinUrl"])

# Let Graph pick a time that suits everyone
suggestions = await graph.calendar.find_times(["alice@example.com", "bob@example.com"],
                                              duration_minutes=30)
best = suggestions[0]["meetingTimeSlot"]

# What is coming up
async for item in graph.calendar.upcoming(days=3):
    print(item["subject"], item["start"]["dateTime"])

# Answer an invitation
await graph.calendar.respond(event_id, "accept", comment="See you there")

Timezone-aware datetimes are sent in UTC; naive ones are read in timezone_name.

6. Files (OneDrive) — graph.files

Works on the signed-in person's OneDrive, or on anyone's when you pass user= (every method takes it; required under application access). Every method takes either a drive path (starts with /, e.g. "/reports/q3.xlsx") or an item id. Names with spaces, # or ? are handled for you.

share_link options

await graph.files.make_folder("/reports/2026", conflict="rename")
item = await graph.files.upload("q3.xlsx", to="/reports/2026/q3.xlsx")

url = await graph.files.share_link("/reports/2026/q3.xlsx", kind="edit")

async for entry in graph.files.folder("/reports/2026"):
    kind = "folder" if "folder" in entry else "file"
    print(kind, entry["name"], entry.get("size"))

async for hit in graph.files.search("budget"):
    print(hit["name"], hit["webUrl"])

await graph.files.download("/reports/2026/q3.xlsx", "downloads/q3.xlsx")
await graph.files.remove("/reports/2026/old.xlsx")

Omit to in upload and the file keeps its own name at the drive root.

7. Teams and chats — graph.teams

Mostly delegated. Under application access, mine(user=) and chats(user=) list a person's teams and chats, but Graph does not let an app post channel or chat messages as someone, so post, reply and send_chat need a signed-in person. Reading channel messages with application permissions is a protected Graph API that Microsoft must approve first.

async for team in graph.teams.mine():
    print(team["id"], team["displayName"])

channel = await graph.teams.channel_by_name(team_id, "deploys")
posted = await graph.teams.post(team_id, channel["id"], "<b>v2.1 is live</b>",
                                html=True, subject="Release", importance="high")

# Replies attach to the thread's first message, so pass the id post() returned
await graph.teams.reply(team_id, channel["id"], posted["id"], "Rollback plan is in the wiki.")

async for chat in graph.teams.chats():
    if chat.get("topic") == "On-call":
        await graph.teams.send_chat(chat["id"], "Paging you now")

Starting a new chat is not wrapped yet; use graph.post("/chats", body=...).

8. People and the directory — graph.users

Works under both access models. Where a method takes user, leave it out for the signed-in person, or pass an id or user principal name. Under application access there is no signed-in person, so pass user.

me = await graph.users.me()
print(me["displayName"], me["jobTitle"])

async for person in graph.users.find("smith"):
    print(person["displayName"], person["mail"])

alice = await graph.users.by_email("alice@example.com")
boss = await graph.users.manager(alice["id"])

async for report in graph.users.reports():
    print("reports to me:", report["displayName"])

await graph.users.photo("me.jpg", size="96x96")

find adds the ConsistencyLevel: eventual header Graph needs for directory search, on every page.

9. Methods every resource has

mail, calendar, files, teams and users all share these, working on the resource's own collection (/me/messages, /me/events, /me/drive/items, /teams, /users). Each takes user= too, which turns /me/... into /users/{user}/... (it has no effect on /teams and /users).

message = await graph.mail.get(message_id, select="subject,body")
await graph.calendar.update(event_id, {"subject": "Moved: Design review"})

people = await graph.users.get_many(["id-1", "id-2", "id-3"], select="displayName,mail")
for result in people:
    if result["status"] == 200:
        print(result["body"]["displayName"])

10. Any other Graph endpoint

Every Graph endpoint is reachable without a dedicated wrapper.

OData options are plain keyword arguments; the $ is added for you.

Any other keyword is sent as a plain query parameter.

Other options

org = await graph.get("/organization", select="displayName,verifiedDomains")

group = await graph.post("/groups", body={
    "displayName": "Project X", "mailNickname": "projectx",
    "mailEnabled": False, "securityEnabled": True,
})

profile = await graph.get("/me/profile", version="beta")

envelope = await graph.request("GET", "/users", top=5, count=True,
                               headers={"ConsistencyLevel": "eventual"})
print(envelope["status"], envelope["body"]["@odata.count"], envelope.get("nextLink"))

Response headers are limited to a safe list: request-id, client-request-id, Date, Retry-After, Content-Type, Location and ETag.

11. Paging

graph.paged(path, **odata) follows @odata.nextLink for you and yields one item at a time. Nothing is held in memory beyond the current page, and you can stop early with break.

count = 0
async for user in graph.paged("/users", select="id,mail", filter="accountEnabled eq true"):
    count += 1
    if count == 1000:
        break

12. Batching

graph.batch(requests) sends many requests with few round-trips. It splits them into groups of 20 (Graph's limit), sends the groups at the same time, and returns results in the order you gave.

Each result has id, status, body and sometimes headers. Failures are reported in place, not raised, so one bad request never loses the others. Only a batch where every group failed raises a GraphError.

results = await graph.batch([
    ("GET", "/me"),
    ("GET", "/me/manager"),
    ("GET", "/me/drive/root/children?$top=5"),
])
for result in results:
    print(result["status"], result["body"])

Batch URLs are relative and have no version prefix. Pass version="beta" to batch to switch all of them.

13. Downloads and uploads

For files outside graph.files, the generic transfer methods take any Graph path.

await graph.download("/me/photo/$value", "me.jpg")
await graph.upload("/sites/SITE_ID/drive/root:/Shared/plan.pdf:/content", "plan.pdf")

14. Errors

Every failure raises one exception type, GraphError, whether it came from Graph, the network or sign-in.

from graphplug import GraphError

try:
    await graph.users.by_email("nobody@example.com")
except GraphError as e:
    if e.code == "itemNotFound":
        print("no such user")
    elif e.status == 429:
        print(f"throttled; retry in {e.retry_after}s")
    else:
        print(f"[{e.status} {e.code}] {e.message} (request {e.request_id})")

Package codes (status is 0)

Throttling (429) and temporary server errors are retried automatically, honouring Retry-After, before an error ever reaches you. Every code and its fix is in the troubleshooting guide.

15. Logging

Off by default. Turn it on with an environment variable:

One JSON line per event, on stderr:

{"level":"info","event":"request","method":"GET","url":"https://graph.microsoft.com/v1.0/users","status":200,"ms":214,"requestId":"a1b2c3d4","errorCode":null}

URLs are logged without their query string, and tokens, secrets, headers and bodies are never logged.

16. Client lifetime and concurrency

import asyncio

async with GraphClient.from_env(max_concurrency=8) as graph:
    ids = ["alice@example.com", "bob@example.com", "carol@example.com"]
    people = await asyncio.gather(*(graph.users.get(i) for i in ids))

More

• USAGE.md: a longer walkthrough
• ARCHITECTURE.md: why it is built the way it is
• Troubleshooting: every error code and its fix
• Samples: runnable scripts

Development

pip install -e .
python -m unittest discover -s tests
python -m build

Releasing

Pushing a version tag publishes to PyPI through GitHub Actions (trusted publishing, no token). Ordinary commits only run the tests.

1. Set __version__ in graphplug/__init__.py (for example "0.2.2") and commit.

2. Tag that commit and push the tag:

   git tag v0.2.2
   git push origin main v0.2.2
   

The workflow tests, builds, checks, and uploads. It refuses to publish if the tag and __version__ differ, because PyPI accepts each version only once.

Licence

MIT